/**
 * Copyright 2008-2009. Chongqing Communications Industry Services Co.,Ltd Information Technology Branch. All rights
 * reserved. <a>http://www.cqcis.com</a>
 */
package com.cqcis.uip.base.framework.service.impl;

import java.io.IOException;
import java.io.InputStream;
import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.util.Collection;
import java.util.Set;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.mina.common.CloseFuture;
import org.apache.mina.common.IdleStatus;
import org.apache.mina.common.IoFilterChain;
import org.apache.mina.common.IoHandler;
import org.apache.mina.common.IoService;
import org.apache.mina.common.IoServiceConfig;
import org.apache.mina.common.IoSession;
import org.apache.mina.common.IoSessionConfig;
import org.apache.mina.common.SessionClosedListener;
import org.apache.mina.common.TrafficMask;
import org.apache.mina.common.TransportType;
import org.apache.mina.common.WriteFuture;

import com.cqcis.uip.base.framework.codec.text.HttpProtocolDecoder;
import com.cqcis.uip.base.framework.common.HttpParam;
import com.cqcis.uip.base.framework.common.InterfaceLog;
import com.cqcis.uip.base.framework.utils.ByteBuffer;
import com.cqcis.uip.base.framework.utils.TimeUtil;

/**
 * 代理InputStream, 加入记录功能
 * 
 * @author huangmy
 * @date   2008-11-3
 */
public class InputStreamRecordBufferAdaptor extends InputStream {
	private static Log log = LogFactory.getLog(InputStreamRecordBufferAdaptor.class);
	private Object finishCheckLock = new Object(); // LOCK
	private ByteBuffer buffer;
	private InputStream is;
	private InterfaceLog intfLog;
	private HttpProtocolDecoder protocolDecoder;

	public InputStreamRecordBufferAdaptor(InputStream is, ByteBuffer buffer, InterfaceLog intfLog) {
		this.is = is;
		this.buffer = buffer;
		this.intfLog = intfLog;
		this.protocolDecoder = new HttpProtocolDecoder();
		this.buffer.setSession(new IoSession() {
			public void addSessionClosedListener(SessionClosedListener l) {
			}
			public Collection<SessionClosedListener> getSessionClosedListeners() {
				return null;
			}
			public CloseFuture close() {
				return null;
			}
			
			public CloseFuture close(boolean closeChannel) {
				return null;
			}

			public boolean containsAttribute(String key) {
				return false;
			}

			public Object getAttachment() {
				return null;
			}

			public Object getAttribute(String key) {
				
				return null;
			}

			public Set<String> getAttributeKeys() {
				
				return null;
			}

			public CloseFuture getCloseFuture() {
				
				return null;
			}

			public IoSessionConfig getConfig() {
				
				return null;
			}

			public long getCreationTime() {
				
				return 0;
			}

			public IoFilterChain getFilterChain() {
				
				return null;
			}

			public IoHandler getHandler() {
				
				return null;
			}

			public int getIdleCount(IdleStatus status) {
				
				return 0;
			}

			public int getIdleTime(IdleStatus status) {
				
				return 0;
			}

			public long getIdleTimeInMillis(IdleStatus status) {
				
				return 0;
			}

			public long getLastIdleTime(IdleStatus status) {
				
				return 0;
			}

			public long getLastIoTime() {
				
				return 0;
			}

			public long getLastReadTime() {
				
				return 0;
			}

			public long getLastWriteTime() {
				
				return 0;
			}

			public SocketAddress getLocalAddress() {
				
				return null;
			}

			public long getReadBytes() {
				
				return 0;
			}

			public long getReadMessages() {
				
				return 0;
			}

			public SocketAddress getRemoteAddress() {
				
				return null;
			}

			public int getScheduledWriteBytes() {
				
				return 0;
			}

			public int getScheduledWriteRequests() {
				
				return 0;
			}

			public IoService getService() {
				
				return null;
			}

			public SocketAddress getServiceAddress() {
				
				return null;
			}

			public IoServiceConfig getServiceConfig() {
				
				return null;
			}

			public TrafficMask getTrafficMask() {
				
				return null;
			}

			public TransportType getTransportType() {
				
				return null;
			}

			public int getWriteTimeout() {
				
				return 0;
			}

			public long getWriteTimeoutInMillis() {
				
				return 0;
			}

			public long getWrittenBytes() {
				
				return 0;
			}

			public long getWrittenMessages() {
				
				return 0;
			}

			public long getWrittenWriteRequests() {
				
				return 0;
			}

			public boolean isClosing() {
				
				return false;
			}

			public boolean isConnected() {
				
				return false;
			}

			public boolean isIdle(IdleStatus status) {
				
				return false;
			}

			public Object removeAttribute(String key) {
				
				return null;
			}

			public void resumeRead() {
			}

			public void resumeWrite() {
			}

			public Object setAttachment(Object attachment) {
				
				return null;
			}

			public Object setAttribute(String key, Object value) {
				
				return null;
			}

			public Object setAttribute(String key) {
				
				return null;
			}

			public void setIdleTime(IdleStatus status, int idleTime) {
			}

			public void setTrafficMask(TrafficMask trafficMask) {
			}

			public void setWriteTimeout(int writeTimeout) {
			}

			public void suspendRead() {
			}

			public void suspendWrite() {
			}

			public WriteFuture write(Object message) {
				return null;
			}
			
		});
	}
	
	private void recordRecvTime() {
		if (this.intfLog.getRecvTime() == null) {
			this.intfLog.setRecvTime(TimeUtil.getCurrentTime());
		}			
	}
	
	@Override
	public int read() throws IOException {
		synchronized (this.finishCheckLock) {
			return doRead();
		}
	}
	
	public int doRead() throws IOException {
		if (isReceivedFinish()) {
			return 0;
		}
		int b = -1;
		
		try {
			b = this.is.read();
		} catch (SocketTimeoutException e) {
			this.intfLog.setRecvRecord(e.getMessage());
			throw e;
		}
	
		if (b >= 0) {
			this.buffer.put((byte)b);
			recordRecvTime();
		}
		return b;
	}
	
	@Override
	public int available() throws IOException {
		return this.is.available();
	}
	
	@Override
	public void close() throws IOException {
		this.is.close();
	}
	
	@Override
	public boolean equals(Object obj) {
		return this.is.equals(obj);
	}
	
	@Override
	public int hashCode() {
		return this.is.hashCode();
	}
	
	@Override
	public synchronized void mark(int readlimit) {
		this.is.mark(readlimit);
	}
	
	@Override
	public boolean markSupported() {
		return this.is.markSupported();
	}
	
	public int read(byte[] b) throws IOException {
		 synchronized (this.finishCheckLock) {
			 return doRead(b);
		 }
	}
	
	public int doRead(byte[] b) throws IOException {
		if (isReceivedFinish()) {
			return 0;
		}
		int count = 0;
		
		try {
			count = this.is.read(b);
		} catch (SocketTimeoutException e) {
			this.intfLog.setRecvRecord(e.getMessage());
			throw e;
		}
	
		if (count > 0) {
			this.buffer.put(b, 0, count);
			recordRecvTime();
		}
		
		return count;
	}
	
	@Override
	public int read(byte[] b, int off, int len) throws IOException {
		synchronized (this.finishCheckLock) {
			return doRead(b, off, len);
		}
	}
	
	public int doRead(byte[] b, int off, int len) throws IOException {
		if (isReceivedFinish()) {
			return 0;
		}
		int count = 0;
		try {
		    count = this.is.read(b, off, len);
		} catch (SocketTimeoutException e) {
			this.intfLog.setRecvRecord(e.getMessage());
			throw e;
		}
				
		if (count > 0) {
			this.buffer.put(b, off, count);
			recordRecvTime();
		}
		
		return count;
	}
	
	private boolean isReceivedFinish() {
	    return isReceivedFinishCheck();
	}
	
	/**
	 * 是否结束结束
	 * 
	 * @return
	 */
	private boolean isReceivedFinishCheck() {		
		int oldpos = this.buffer.position();
		int oldLimit = this.buffer.limit();
				
		if (oldpos <= 0) {
			return false;
		}
				
		try {
			this.buffer.flip();
			HttpParam p = (HttpParam)this.protocolDecoder.decode(this.buffer);
			if (p != null) {
				if (log.isDebugEnabled()) {
					log.debug("Recv HttpParam: " + p.toString());
				}
				
				return true;
			}
			
			if (log.isDebugEnabled()) {
				this.buffer.position(0);
				log.debug("the HTTP is not finished, http is: " +  new String(this.buffer.getPrintArray()));
			}
		} catch (Exception e) {
			log.error("Catch Exception in Check HTTP.", e);
			return false;
		} finally {
			this.buffer.position(oldpos);
			this.buffer.limit(oldLimit);
		}
						
		return false;
	}
	
	@Override
	public synchronized void reset() throws IOException {
		this.is.reset();
	}
	
	@Override
	public long skip(long n) throws IOException {
		return this.is.skip(n);
	}
	
	@Override
	public String toString() {
		return this.is.toString();
	}
}
